package com.tid.hbasees.repo;


import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.tid.hbasees.model.Device;
import com.tid.hbasees.utils.JsonHelper;
import com.tid.hbasees.utils.hbase.HBaseHelper;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Map;

@Component
public class DeviceRepo {
    private static final Logger LOG = Logger.getLogger(DeviceRepo.class);

    public static String TABLE_NAME = "device";
    public static String INDEX_TABLE_NAME = TABLE_NAME + "_index";

    @Resource(name = "hbaseHelper")
    private HBaseHelper hbaseHelper;

    @Resource
    private JsonHelper jsonHelper;

    private HTableInterface getTable() throws IOException {
        return getTable(TABLE_NAME);
    }

    private HTableInterface getTable(String tableName) throws IOException {
        return hbaseHelper.getTable(tableName);
    }

    private List<Device> getDevicesFromResult(ResultScanner rs) {
        List<Device> result = Lists.newLinkedList();
        for (Result r : rs) {
            String json = hbaseHelper.getJsonContent(r);
            if (json != null) {
                Optional<Device> nodeOptional = jsonHelper.readValue(json, Device.class);
                if (nodeOptional.isPresent()) {
                    result.add(nodeOptional.get());
                }
            }
        }
        return result;
    }

    public List<Device> getByType(String deviceTypeId, long startTime, long endTime) {
        Scan s = hbaseHelper.buildTypicalScan(deviceTypeId, startTime, endTime);

        try (HTableInterface table = getTable(); ResultScanner rs = table.getScanner(s)) {

            return getDevicesFromResult(rs);

        } catch (Exception e) {
            e.printStackTrace();
            LOG.error("failed to get nodes with condition: " + startTime + " " + endTime + " " + deviceTypeId);
            return Lists.newLinkedList();
        }
    }

    public List<Device> getByNodeId(String nodeId, long startTime, long endTime) {
        Map<String, Scan> scans = Maps.newHashMap();
        scans.put(INDEX_TABLE_NAME, hbaseHelper.buildTypicalScan(nodeId, startTime, endTime));
        try {
            return hbaseHelper.multiScan(scans, TABLE_NAME, Device.class);
        } catch (Exception e) {
            e.printStackTrace();
            return Lists.newLinkedList();
        }
    }

    public void save(Device device) {

    }

    public void save(List<Device> devices) {

    }
}
